package core

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"gark-elastic-job/common"
	"gark-elastic-job/engine"
	logger "github.com/sirupsen/logrus"
	"go.etcd.io/etcd/clientv3"
	"math/rand"
	"sync"
	"time"
)

// 工作群组管理者
type GroupManager struct {
	node   *JobNode
	groups map[string]*Group
	locker *sync.RWMutex
}

func NewGroupManager(node *JobNode) *GroupManager {
	logger.Debug("start Group Manager")
	groupManager := &GroupManager{
		node:   node,
		groups: make(map[string]*Group),
		locker: &sync.RWMutex{},
	}
	go groupManager.loopLoadGroups()
	go groupManager.watchGroupPath()
	return groupManager
}

func (groupManager *GroupManager) selectClient(name string) (*Client, error) {
	if group, ok := groupManager.groups[GroupConfPath+name]; ok {
		return group.selectClient()
	}
	return nil, errors.New("group doesn't exist")
}

func (groupManager *GroupManager) addGroup(name, path string, node *JobNode) {
	groupManager.locker.Lock()
	defer groupManager.locker.Unlock()

	if _, ok := groupManager.groups[path]; ok {
		return
	}

	group := newGroup(name, path, node)
	groupManager.groups[group.Path] = group
	logger.Infof("success add group: %s, path: %s to group manager", name, path)
}

func (groupManager *GroupManager) loopLoadGroups() {
	keys, valResp, err := groupManager.node.etcd.GetWithPrefixKey(GroupConfPath)
	if err != nil {
		logger.Errorf("failed to get groups, err: %v", err)
		return
	}

	if len(keys) == 0 {
		logger.Warnf("no group to be load")
		return
	}

	for _, val := range valResp {
		groupConf := &engine.GroupConf{}
		err := json.Unmarshal(val, groupConf)
		if err != nil {
			logger.Warnf("failed to decode group conf, err: %v", err)
			continue
		}
		groupManager.addGroup(groupConf.Name, GroupConfPath+groupConf.Name, groupManager.node)
	}

}

func (groupManager *GroupManager) watchGroupPath() {
	keyChangeResponse := groupManager.node.etcd.WatchKeyWithPrefix(GroupConfPath)
	for event := range keyChangeResponse.Event {
		groupManager.handleGroupChangeEvent(event)
	}
}

func (groupManager *GroupManager) handleGroupChangeEvent(event *engine.KeyChangeEvent) {
	switch event.Type {
	case engine.KeyCreateChangeEvent:
		groupManager.handleGroupCreateEvent(event.Value)
	case engine.KeyUpdateChangeEvent:
		//ignore
	case engine.KeyDeleteChangeEvent:
		groupManager.handleGroupDeleteEvent(event.Key)
	}
}

func (groupManager *GroupManager) handleGroupCreateEvent(value []byte) {
	groupConf := &engine.GroupConf{}
	err := json.Unmarshal(value, groupConf)
	if err != nil {
		logger.Warnf("decode err: %v", err)
		return
	}
	groupManager.addGroup(groupConf.Name, GroupConfPath+groupConf.Name, groupManager.node)
}

func (groupManager *GroupManager) handleGroupDeleteEvent(key string) {

}

// 工作群组
type Group struct {
	Name       string
	Path       string
	node       *JobNode
	watchPath  string
	locker     *sync.RWMutex
	clients    map[string]*Client
	watcher    clientv3.Watcher
	cancelFunc context.CancelFunc
}

func newGroup(name, path string, node *JobNode) *Group {
	group := &Group{
		Name:      name,
		Path:      path,
		node:      node,
		watchPath: fmt.Sprintf(ClientPath, name),
		locker:    &sync.RWMutex{},
		clients:   make(map[string]*Client),
	}

	go group.watchClientPath()
	go group.loopLoadClient()

	return group
}

func (g *Group) selectClient() (*Client, error) {
	g.locker.RLock()
	defer g.locker.RUnlock()

	size := len(g.clients)
	if size == 0 {
		return &Client{}, errors.New(fmt.Sprintf("group: %v has no avaliable client to select", g.Name))
	}
	// TODO 这里可以考虑使用接口选择不同的负载均衡算法
	target := rand.Intn(size)
	index := 0
	for _, c := range g.clients {
		if index == target {
			return c, nil
		}
		index++
	}
	return &Client{}, errors.New("something went wrong to select client")
}

func (g *Group) watchClientPath() {
	keyChangeResponse := g.node.etcd.WatchKeyWithPrefix(g.watchPath)
	g.watcher = keyChangeResponse.Watcher
	g.cancelFunc = keyChangeResponse.CancelFunc
	for event := range keyChangeResponse.Event {
		go g.handleClientChangeEvent(event)
	}
}

func (g *Group) handleClientChangeEvent(event *engine.KeyChangeEvent) {
	switch event.Type {
	case engine.KeyCreateChangeEvent:
		g.handleClientCreateEvent(event.Key, string(event.Value))
	case engine.KeyUpdateChangeEvent:
		// todo - ignore now
	case engine.KeyDeleteChangeEvent:
		g.handleClientDeleteEvent(event.Key)
	}
}

func (g *Group) handleClientCreateEvent(path, name string) {
	g.addClient(name, path)
}

func (g *Group) loopLoadClient() {
RETRY:
	prefix := fmt.Sprintf(ClientPath, g.Name)
	keys, values, err := g.node.etcd.GetWithPrefixKey(prefix)
	if err != nil {
		logger.Warnf("failed to load for the group: %s, err: %v", g.Name, err)
		time.Sleep(time.Second)
		goto RETRY
	}

	if len(keys) == 0 {
		logger.Debugf(" no client to be load for the group: %s", g.Name)
		time.Sleep(time.Second)
		goto RETRY
	}

	clientCounts := len(values)
	for i := 0; i < clientCounts; i++ {
		key := string(keys[i])
		value := string(values[i])
		if value == "" {
			logger.Warnf("the client value is nil for path:%s", key)
			continue
		}
		g.addClient(value, key)
	}

}

func (g *Group) addClient(value, path string) {
	g.locker.Lock()
	defer g.locker.Unlock()

	clientMetaData := &common.ClientMetaData{}
	err := json.Unmarshal([]byte(value), clientMetaData)
	if err != nil {
		return
	}
	if _, ok := g.clients[path]; ok {
		logger.Infof("[group]client %v already exists", value)
		return
	}

	g.clients[path] = &Client{
		ID:   clientMetaData.ID,
		IP:   clientMetaData.IP,
		Path: path,
	}
	logger.Debugf("group: %s add client: %v for the path: %s", g.Name, clientMetaData, path)
}

func (g *Group) handleClientDeleteEvent(path string) {
	g.locker.Lock()
	defer g.locker.Unlock()

	if _, ok := g.clients[path]; ok {
		logger.Debugf("group: %s remove client for the path: %s", g.Name, path)
		delete(g.clients, path)
		return
	}
}

type Client struct {
	ID   string
	IP   string
	Path string
}
